package com.ruyuan.ingestion.sink;

import com.ruyuan.ingestion.common.Parsable;
import com.ruyuan.ingestion.common.Persistable;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import java.sql.CallableStatement;
import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Properties;

public class MysqlWriter implements Persistable {

    private String jdbcURL = null;

    private String user = null;

    private String password = null;

    private String table = null;

    private Parsable<String> parser = null;


    @Override
    public void initialize(Properties properties) {
        this.table = table;
        this.parser = parser;
    }

    /**
     * 数据写入mysql
     * */
    @Override
    public int write(ConsumerRecords<String, String> records) throws Exception {
        int numCount =0;
        if (this.jdbcURL == null || this.jdbcURL.isEmpty()) {
            throw new Exception("jdbcURL is not initialized");
        }

        Connection connection = DriverManager.getConnection(this.jdbcURL, this.user, this.password);
        try {
            long passHead = 0;
            for(ConsumerRecord<String,String> record:records) {
                String[] element = record.value().split(",", -1);
                if (passHead ==0 && this.parser.isHeader(element)) {
                    passHead = 1;
                    continue;
                }

                if (this.parser.isValid(element)) {
                    CallableStatement callableStatement = connection.prepareCall(this.parser.parse(element));
                    try {
                        callableStatement.execute();
                        numCount++;
                    }finally {
                        callableStatement.close();
                    }
                }
            }
        }finally {
            connection.close();
        }
        return numCount;
    }
}
